package com.atguigu.actual.edu0417.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.actual.edu0417.app.BaseApp;
import com.atguigu.actual.edu0417.beans.CartAddUuBean;
import com.atguigu.actual.edu0417.func.BeanToJsonStrMapFcuntion;
import com.atguigu.actual.edu0417.unils.DateFormatUtil;
import com.atguigu.actual.edu0417.unils.DorisSinkUtil;
import com.atguigu.actual.edu0417.unils.DorisUtil;
import com.atguigu.actual.edu0417.unils.MyKafkaUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * @author: 洛尘
 * @since: 2023-10-21 15:29
 * @description: 加购独立用户数统计
 **/
public class DwsTradeCartAddUuWindow extends BaseApp {
    //基本环境准备
    public static void main(String[] args) {
        DwsTradeCartAddUuWindow dwsTradeCartAddUuWindow = new DwsTradeCartAddUuWindow();
        dwsTradeCartAddUuWindow.base(args);
    }
        //处理数据
    @Override
    public void process(StreamExecutionEnvironment env) {
        //创建消费者，消费dwd数据
        String topic="dwd_trade_cart_add";
        String groupID="dws_trade_cart_group";

        KafkaSource<String> kafkaSource = MyKafkaUtil.getKafkaSource(topic, groupID);

        SingleOutputStreamOperator<String> kafkaDS = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka_cart_source");
        //基本数据转换
        SingleOutputStreamOperator<JSONObject> jsonObj = kafkaDS.map(
                new MapFunction<String, JSONObject>() {
                    @Override
                    public JSONObject map(String jsonStr) throws Exception {
                        JSONObject jsonObject = JSON.parseObject(jsonStr);
                        return jsonObject;

                    }
                }
        );
        //指定WaterMark并且提取事件时间字段
        SingleOutputStreamOperator<JSONObject> waterMarkDS = jsonObj.assignTimestampsAndWatermarks(
                WatermarkStrategy.<JSONObject>forMonotonousTimestamps()
                        .withTimestampAssigner(
                new SerializableTimestampAssigner<JSONObject>() {
                    @Override
                    public long extractTimestamp(JSONObject element, long recordTimestamp) {
                        return element.getLong("ts") * 1000;
                    }
                }
        ));
        //统计加购独立用户数量，所以用user_id,keyby
        // id |     user_id |       course_id |     course_name |    cart_price |   deleted |    sold |     ts
        //12428           ,180            ,60,JavaScript            ,200               ,0             ,0    ,1697790515
        KeyedStream<JSONObject, String> keyedDS = waterMarkDS.keyBy(user -> user.getString("user_id"));
        //状态编程 判断是否为独立用户
        SingleOutputStreamOperator<JSONObject> processDS = keyedDS.process(
                new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                    private ValueState<String> lastCartDateState;

                    @Override
                    public void open(Configuration parameters) throws Exception {

                        ValueStateDescriptor<String> valueCartDateDec = new ValueStateDescriptor<>("valueCartDateDec", String.class);
                        //设置状态失效时间的原因是因为统计的指标是当天的独立加购用户，所以这个用户的信息存在时间就是24小时，那么状态中存放该用户信息的时间是有时效的
                        valueCartDateDec.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());
                        lastCartDateState = getRuntimeContext().getState(valueCartDateDec);
                    }

                    @Override
                    public void processElement(JSONObject jsonObject, Context ctx, Collector<JSONObject> out) throws Exception {
                        String lastDate = lastCartDateState.value();
                        Long ts = jsonObject.getLong("ts") * 1000;
                        String curDate = DateFormatUtil.toDate(ts);
                        if (StringUtils.isEmpty(lastDate) || !lastDate.equals(curDate)) {
                            out.collect(jsonObject);
                            lastCartDateState.update(curDate);
                        }
                    }
                }
        );
        //判断结束，开窗聚合计算
        AllWindowedStream<JSONObject, TimeWindow> windowDS
                = processDS.windowAll(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)));
        //聚合
        SingleOutputStreamOperator<CartAddUuBean> aggregateDS = windowDS.aggregate(
                new AggregateFunction<JSONObject, Long, Long>() {
                    @Override
                    public Long createAccumulator() {
                        return 0L;
                    }

                    @Override
                    public Long add(JSONObject value, Long accumulator) {
                        return ++accumulator;
                    }

                    @Override
                    public Long getResult(Long accumulator) {
                        return accumulator;
                    }

                    @Override
                    public Long merge(Long a, Long b) {
                        return null;
                    }
                },
                new AllWindowFunction<Long, CartAddUuBean, TimeWindow>(){
                    @Override
                    public void apply(TimeWindow window, Iterable<Long> values, Collector<CartAddUuBean> out) throws Exception {
                        String stt = DateFormatUtil.toYmdHms(window.getStart());
                        String edt = DateFormatUtil.toYmdHms(window.getEnd());
                        String curDate = DateFormatUtil.toDate(window.getStart());
                        for (Long value : values) {
                            out.collect(new CartAddUuBean(
                                    stt,edt,curDate,value
                            ));
                        }
                    }
                }
        );
        //写入doris
        aggregateDS.print(">>>");
        aggregateDS
                .addSink(DorisSinkUtil.<CartAddUuBean>getJdbcSink("insert into dws_trade_cart_add_uu_window values(?,?,?,?)"));
    }
}